package com.zyl.mq.util;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @author zyl
 * @Description
 * @since 2023/11/24 16:59
 */

public class RabbitMQSubscriber {
    private ConnectionFactory factory;
    private Connection connection;
    private Channel channel;
    private String queueName;

    public RabbitMQSubscriber(String host, String queueName, String username, String password) {
        // 创建 RabbitMQ 连接工厂
        this.factory = new ConnectionFactory();
        this.factory.setHost(host);
        this.factory.setUsername(username);
        this.factory.setPassword(password);
        this.queueName = queueName;
    }

    /**
     * 启动消息监听
     * @param messageHandler
     * @throws IOException
     */
    public void startListening(MessageHandler messageHandler) throws IOException {
        try {
            // 创建 RabbitMQ 连接
            connection = factory.newConnection();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        // 创建消息通道
        channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(queueName, false, false, false, null);
        // 设置消息的传递回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            // 处理接收到的消息
            String message = new String(delivery.getBody(), "UTF-8");
            messageHandler.handleMessage(message);
        };
        // 订阅队列，并设置消息传递回调
        channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
        });
    }

    /**
     * 关闭链接
     * @throws IOException
     * @throws TimeoutException
     */
    public void close() throws IOException, TimeoutException {
        if (connection != null && connection.isOpen()) {
            connection.close();
        }
    }

    /**
     * 自定义处理接收到的消息
     */
    public interface MessageHandler {
        void handleMessage(String message);
    }
}
